/*
* GPL
* Free soft,no warranted
* We are all in the mud, but some of us are looking at the stars.
* We are not born for ourselves.
*/
/**
* @file io_service_pool.hpp
* @Synopsis
* @author Chai JunJun
* @date 2009-07-10
*/

#ifndef QSERV_IO_SERVICE_POOL_HPP__
#define QSERV_IO_SERVICE_POOL_HPP__

#include <boost/asio.hpp>
#include <boost/assert.hpp>
#include <boost/thread.hpp>
#include <boost/noncopyable.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/bind.hpp>
#include <vector>

namespace qserv 
{

    /// A pool of io_service objects.
    class io_service_pool
        : private boost::noncopyable
    {
    public:
        /// Construct the io_service pool.
        explicit io_service_pool(std::size_t pool_size)
            : m_ioservices(),
            m_works(),
            m_threads(),
            m_next_ioservice(0),
            m_block(false)
        {
            BOOST_ASSERT(pool_size != 0);

            // Create io_service pool.
            for (std::size_t i = 0; i < pool_size; ++i)
            {
                io_service_ptr io_service(new boost::asio::io_service);
                m_ioservices.push_back(io_service);
            }
        }

        /// Destruct the pool object.
        ~io_service_pool()
        {
            stop();

            // Destroy all work.
            for (std::size_t i = 0; i < m_works.size(); ++i)
                m_works[i].reset();
            m_works.clear();

            // Destroy io_service pool.
            for (std::size_t i = 0; i < m_ioservices.size(); ++i)
                m_ioservices[i].reset();
            m_ioservices.clear();
        }

        std::size_t size()
        {
            return m_ioservices.size();
        }

        /// Start all io_service objects in nonblock model.
        void start()
        {
            start(false);
        }

        /// Run all io_service objects in block model.
        void run()
        {
            start(true);
        }

        /// Stop all io_service objects in the pool.
        void stop()
        {
            if (m_threads.size() == 0)
                return;

            // Allow all operations and handlers to be finished normally,
            // the work object may be explicitly destroyed.

            // Destroy all work.
            for (std::size_t i = 0; i < m_works.size(); ++i)
                m_works[i].reset();
            m_works.clear();

            if (!m_block)
                wait();
        }

        /// Get an io_service to use.
        boost::asio::io_service& get_io_service()
        {
            boost::asio::io_service& io_service = *m_ioservices[m_next_ioservice];
            if (++m_next_ioservice == m_ioservices.size())
                m_next_ioservice = 0;
            return io_service;
        }

        /// Get the certain io_service to use.
        boost::asio::io_service& get_io_service(std::size_t offset)
        {
            BOOST_ASSERT(offset < m_ioservices.size());
            return *m_ioservices[offset];
        }

    private:
        /// Wait for all threads in the pool to exit.
        void wait()
        {
            if (m_threads.size() == 0)
                return;

            // Wait for all threads in the pool to exit.
            for (std::size_t i = 0; i < m_threads.size(); ++i)
            {
                m_threads[i]->join();
                std::cout << "thread is out\n";
            }
            // Destroy all threads.
            m_threads.clear();
        }

        /// Start all io_service objects in the pool.
        void start(bool block)
        {
            if (m_threads.size() != 0)
                return;

            // Reset the io_service in preparation for a subsequent run() invocation.
            std::vector<io_service_ptr>::size_type pool_size = m_ioservices.size();
            for (std::size_t i = 0; i < pool_size; ++i)
            {
                m_ioservices[i]->reset();
            }

            for (std::size_t i = 0; i < pool_size; ++i)
            {
                // Give the io_service work to do so that its run() functions will not
                // exit until work was explicitly destroyed.
                work_ptr work(new boost::asio::io_service::work(*m_ioservices[i]));
                m_works.push_back(work);

                // Create a thread to run the io_service.
                thread_ptr thread(new boost::thread(
                    boost::bind(&boost::asio::io_service::run, m_ioservices[i])));
                m_threads.push_back(thread);
            }

            m_block = block;

            if (block)
                wait();
        }

    private:
        typedef boost::shared_ptr<boost::asio::io_service> io_service_ptr;
        typedef boost::shared_ptr<boost::asio::io_service::work> work_ptr;
        typedef boost::shared_ptr<boost::thread> thread_ptr;

        /// The pool of io_services.
        std::vector<io_service_ptr> m_ioservices;

        /// The work that keeps the io_services running.
        std::vector<work_ptr> m_works;

        /// The pool of threads for running individual io_service.
        std::vector<thread_ptr> m_threads;

        /// The next io_service to use for a connection.
        std::size_t m_next_ioservice;

        /// Flag to indicate that start() functions will block or not.
        bool m_block;
    };

    typedef boost::shared_ptr<io_service_pool> IOServicePoolPtr;

    ///helper class of io_service objects pool
    ///Dual_lock singleton
    class IOServicePoolSingleton
    {
    private:
        IOServicePoolSingleton()
        {

        }
        
    public:
        ~IOServicePoolSingleton()
        {

        }

        static IOServicePoolPtr getSingleton(std::size_t pool_size);
    private:
        static IOServicePoolPtr m_iosptr;
        static boost::recursive_mutex m_mutex;
    };
} // namespace qserv

#endif // QSERV_IO_SERVICE_POOL_HPP__
